from pyspark.context import SparkContext

# 1、创建spark的执行环境
# master: 执行运行模式
# appName： 任务名称
sc = SparkContext(master="local", appName="WordCount")


# 2、读取数据
# RDD:弹性的分布式数据集
# 可以读取HDFS中的数据
lines_rdd = sc.textFile("../../data/words")

print(f"lines_rdd分区数：{lines_rdd.getNumPartitions()}")

# 3、一行转换成多行
# flatMap:相当于sql中的Explode
words_rdd = lines_rdd.flatMap(lambda line: line.split(","))

print(f"words_rdd分区数：{words_rdd.getNumPartitions()}")

# 4、转换成kv格式
# map: 一行一行处理数据
kv_rdd = words_rdd.map(lambda word: (word, 1))

print(f"kv_rdd分区数：{kv_rdd.getNumPartitions()}")

# 5、统计单词的数量
# reduceByKey： 对相同key的value进行聚合计算
# 经过会产生shuffle的算子之后分区数可以改，默认和前面的rdd保持一致,  设置多少分区由数据量决定
# 分区太多会产生小文件，分区太少计算速度慢
count_rdd = kv_rdd.reduceByKey(lambda x, y: x + y, numPartitions=10)

print(f"count_rdd分区数：{count_rdd.getNumPartitions()}")

# 整理结果
result_rdd = count_rdd.map(lambda kv: f"{kv[0]}\t{kv[1]}")

# 保存结果
result_rdd.saveAsTextFile("../../data/word_count")

# http://localhost:4040
while True:
    pass

